package com.edu.learning.concurrent.sample.mode4;

import java.util.Random;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import static java.util.concurrent.TimeUnit.SECONDS;

/**
 * SynchronousQueue是一个比较特殊的阻塞队列，它具有以下几个特点：
 * 1. 一个调用插入方法的线程必须等待另一个线程调用取出方法；
 * 2. 队列没有容量Capacity（或者说容量为0），事实上队列中并不存储元素，它只是提供两个线程进行信息交换的场所；
 * 3. 由于以上原因，队列在很多场合表现的像一个空队列。不能对元素进行迭代，不能peek元素，poll会返回null；
 * 4. 队列中不允许存入null元素。
 * 5. SynchronousQueue如同ArrayedBlockingQueue一样，支持“公平”策略，在构造函数中可以传入false或true表示是否支持该策略。
 *
 * 例子，5个Producer产生产品，存入队列；5个Consumer从队列中取出产品，进行消费。
 *
 * @author ll
 * @Date 2017/10/26 11:25
 */
public class SynchronizeQueueDemo {
    public static void main(String[] args) {
        SynchronousQueue<String> synchronousQueue = new SynchronousQueue<>(false);
        ExecutorService executorService = new ThreadPoolExecutor(
                32,
                32,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(1024),
                new ThreadPoolExecutor.CallerRunsPolicy());
        for (int i = 0; i < 5; i ++) {
            executorService.submit(new Producer("Producer " + i, synchronousQueue));
        }

        for (int i = 0; i < 5; i ++) {
            executorService.submit(new Consumer("Consumer " + i, synchronousQueue));
        }
        executorService.shutdown();
    }

    static class Producer implements Runnable {

        private final SynchronousQueue<String> synchronousQueue;
        private final String name;
        private static Random rand = new Random(47);
        private static AtomicInteger productID = new AtomicInteger(0);

        public Producer(String name, SynchronousQueue<String> synchronousQueue) {
            this.name = name;
            this.synchronousQueue = synchronousQueue;
        }

        @Override
        public void run() {
            try {
                StringBuilder sb = new StringBuilder();
                for (int i = 0; i < 5; i ++) {
                    SECONDS.sleep(rand.nextInt(5));
                    String str = "Product" + productID.getAndIncrement();
                    synchronousQueue.put(str);
                    sb.append(name).append(" product ").append(str).append(", queue.size = ").append(synchronousQueue.size());
                    System.out.println(sb);//name + " product " + sb + "\t , queue.size = " + blockingQueue.size()
                }
                System.out.println(name + " is over");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    static class Consumer implements Runnable {

        private final SynchronousQueue<String> synchronousQueue;
        private final String name;

        public Consumer(String name, SynchronousQueue<String> synchronousQueue) {
            this.name = name;
            this.synchronousQueue = synchronousQueue;
        }

        @Override
        public void run() {
            try {
                StringBuilder sb = new StringBuilder();
                for (int i = 0; i < 5; i ++) {
                    String str = synchronousQueue.take();
                    sb.append(name).append(" consume ").append(str).append(", queue size = ").append(synchronousQueue.size());
                    System.out.println(sb);
                }
                System.out.println(name + " is over");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
